package NBA;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class NBAClear {
	
	public static class MyMapper extends Mapper<LongWritable, Text, NullWritable, Text> {		
		Map<String, String> nameMap = new HashMap<String, String>();  // 用于保存新旧队名映射
		Map<String, List<String>> areaMap = new HashMap<String, List<String>>();  // 用于映射东西区球队
		Text newValue = new Text();	
		
		@Override
		protected void setup(Mapper<LongWritable, Text, NullWritable, Text>.Context context) throws IOException, InterruptedException{
			
			// 1.新旧队名映射
			List<String> mapList = getFiletoList("A:/input/mappings.txt");
			for (String s : mapList) {
				String[] result = s.split(",");
				nameMap.put(result[0],result[1]);
			}	
			// 2.东区球队
			List<String> eastList = getFiletoList("A:/input/east.txt");
			areaMap.put("east", eastList);	
			//3.西区球队
			List<String> westList = getFiletoList("A:/input/west.txt");
			areaMap.put("west", westList);		
		}

		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{
			String[] split = value.toString().split(",");  // 使用“，”拆分
			// 获取第一和第二
			String Champion = split[2];
			String Second = split[4];
			// 新旧队名替换
			String newName = nameMap.get(Champion);
			Champion = newName == null ? Champion : newName;
			String newName2 = nameMap.get(Second);
			Second = newName2 == null ? Second : newName2;
			split[2] = Champion;
			split[4] = Second;
			int year = Integer.parseInt(split[0]);
			// 设置分区
			if (year >= 1970) {
				String areaFlag = "";
				List<String> eastList = areaMap.get("east");
				List<String> westList = areaMap.get("west");
				if(eastList.contains(Champion))
					areaFlag = "E";
				else if (westList.contains(Champion)) 
					areaFlag = "W";
				newValue.set(split[0] + "," + split[1] + "," + split[2] + "," + split[3] + "," + split[4] + "," + areaFlag);
				
			}else {
				newValue.set(split[0] + "," + split[1] + "," + split[2] + "," + split[3] + "," + split[4]);
			}
			context.write(NullWritable.get(), newValue);
		}		
	}
	
	// 按行读取文件
	public static List getFiletoList(String name) {
		// 使用ArrayList来存储每行读取到的字符串
		List<String> List = new ArrayList<String>();
		try {
			FileReader fr = new FileReader(name);
			BufferedReader bf = new BufferedReader(fr);
			String str;
			// 按行读取字符串
			while ((str = bf.readLine()) != null) {
				List.add(str);
			}
			bf.close();
			fr.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return List;
	}
	
	// 主方法
	public static void main(String[] args) throws Exception{
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		
		job.setJarByClass(NBAClear.class);
		job.setMapperClass(MyMapper.class);
		
		job.setMapOutputKeyClass(NullWritable.class);
		job.setMapOutputValueClass(Text.class);
		
		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(Text.class);
		
		job.setNumReduceTasks(0);  // 没有reduce时MR会自动提供一个
		
		String input = "A:/input/nba.csv";
		String output = "A:/output4";
		
		FileInputFormat.setInputPaths(job,new Path(input));
		FileOutputFormat.setOutputPath(job,new Path(output));
		
		if(job.waitForCompletion(true)) {
			System.out.println("ִ运行成功!");
			System.exit(0);
		}else {
			System.out.println("ִ运行失败...");
			System.exit(1);
		}
	}
	
	
}
